package cn.com.tt.flink.sql.engine.processor;

import cn.com.tt.flink.runner.AbstractProcessor;
import cn.com.tt.flink.sql.engine.common.SqlTypeConstants;
import cn.com.tt.flink.sql.engine.pojo.JobConfig;
import cn.com.tt.flink.sql.engine.pojo.JobSql;
import cn.com.tt.flink.sql.engine.pojo.sql.AbstractBmFlinkSql;
import cn.com.tt.flink.sql.engine.pojo.sql.BmDdlSql;
import cn.com.tt.flink.sql.engine.pojo.sql.BmInsertSql;
import cn.com.tt.flink.sql.engine.pojo.sql.BmPrintSql;
import cn.com.tt.flink.sql.engine.utils.LambdaUtils;
import cn.com.bluemoon.bd.utils.RegexUtils;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class SqlProcessor extends AbstractProcessor {

    private static final String SQL_DELIMITER = ";";

    public SqlProcessor(StreamTableEnvironment tableEnv) {
        super(tableEnv);
    }


    @Override
    public void process() throws Exception {
        JobSql jobSql = JobConfig.getInstance().getSqlStatement();
        List<AbstractBmFlinkSql> sqlList = parseSql(jobSql);
        execute(sqlList);
    }

    private void execute(List<AbstractBmFlinkSql> sqlList) {
        //insert语句组装成set一次执行，避免一条语句提交一个作业
        StatementSet insertSqlStatementSet = tableEnv.createStatementSet();

        //将insert sql 添加到set,并执行非insert sql,
        sqlList.forEach(sql -> {

            if (sql instanceof BmInsertSql) {

                try {
                    insertSqlStatementSet.addInsertSql(sql.getSqlStatement());
                } catch (Exception e) {
                    throw new RuntimeException("sql执行失败：" + sql.getSqlStatement(), e);
                }
            } else if (sql instanceof BmPrintSql) {
                LambdaUtils.consumerWrapper(AbstractBmFlinkSql::process).accept(sql);
                String insertIntSql = null;
                try {
                    insertIntSql = ((BmPrintSql) sql).getInsertIntSql();
                    insertSqlStatementSet.addInsertSql(insertIntSql);
                } catch (Exception e) {
                    throw new RuntimeException("sql执行失败：" + insertIntSql, e);
                }
            } else {
                LambdaUtils.consumerWrapper(AbstractBmFlinkSql::process).accept(sql);
            }
        });

        //执行insert sql set
        insertSqlStatementSet.execute();
    }

    private List<AbstractBmFlinkSql> parseSql(JobSql jobSql) {
        //移除注释行
        String sqlStatement = removeAnnotation(jobSql.getSqlStatement());

        //用定义好的分隔符(;号)切分成一段段sql;
        String[] sqlArray = sqlStatement.split(SQL_DELIMITER);

        return Arrays.stream(sqlArray).
                map(sql -> replaceVariable(sql, jobSql.getLocalParamMap()))
                .map(this::createSql)
                .collect(Collectors.toList());
    }

    private String replaceVariable(String sql, Map<String, String> localParamMap) {
        for (Map.Entry<String, String> localParam : localParamMap.entrySet()) {
            sql = sql.replace("${" + localParam.getKey() + "}", localParam.getValue());
        }
        return sql;
    }

    /**
     * sql语句移除注释行
     *
     * @param sqlStatement
     * @return
     */
    private String removeAnnotation(String sqlStatement) {
        sqlStatement = sqlStatement.trim();

        //去掉以--开头的行
        sqlStatement = Arrays.stream(sqlStatement.split("\n"))
                .filter(s -> !(s.trim().startsWith("--")))
                .reduce("", (a, b) -> a + b + '\n');

        //去掉最后的空行
        return sqlStatement.trim();
    }

    /**
     * 构造不同类型的sql
     *
     * @param sqlstatement
     * @return
     */
    private AbstractBmFlinkSql createSql(String sqlstatement) {
        int sqlType = getSqlType(sqlstatement);
        switch (sqlType) {
            case SqlTypeConstants.DDL_SQL:
                return new BmDdlSql(sqlstatement, tableEnv);
            case SqlTypeConstants.INSERT_SQL:
                return new BmInsertSql(sqlstatement, tableEnv);
            case SqlTypeConstants.CREATE_CATALOG_SQL:
                return new BmInsertSql(sqlstatement, tableEnv);
            case SqlTypeConstants.USE_CATALOG_SQL:
                return new BmInsertSql(sqlstatement, tableEnv);
            case SqlTypeConstants.USE_DIALECT_SQL:
                return new BmInsertSql(sqlstatement, tableEnv);
            case SqlTypeConstants.PRINT_SQL:
                String printText = getPrintText(sqlstatement);
                String sql = sqlstatement.replace(printText, "").trim();
                return new BmPrintSql(printText, sql, tableEnv);
            default:
                return new BmDdlSql(sqlstatement, tableEnv);
        }
    }

    private int getSqlType(String sql) throws RuntimeException {
        String trimSql = sql.trim().toLowerCase().replaceAll("\\s+", " ");

        if (trimSql.startsWith("insert")) {
            return SqlTypeConstants.INSERT_SQL;
        } else if (trimSql.startsWith("set table.sql-dialect")) {
            return SqlTypeConstants.USE_DIALECT_SQL;
        } else if (trimSql.startsWith("set table.sql-catalog")) {
            return SqlTypeConstants.USE_CATALOG_SQL;
        } else if (trimSql.startsWith("create catalog")) {
            return SqlTypeConstants.CREATE_CATALOG_SQL;
        } else if (trimSql.startsWith("@print")) {
            return SqlTypeConstants.PRINT_SQL;
        } else {
            return SqlTypeConstants.DDL_SQL;
        }
    }

    private String getPrintText(String sqlStatement) {
        String trimSql = sqlStatement.replaceAll("\\s+", " ").trim();
        String p;
        if (trimSql.charAt(6) == '(') {
            p = "(@print.*?\\))";
        } else {
            p = "(@print.*?) ";
        }

        return RegexUtils.extractData(trimSql, p);
    }


}
